package com.saga.energy.sink;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.saga.energy.bean.UserToSpace;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.List;

public class ZJSpaceToMysql extends RichSinkFunction<JSONArray> {

    private PreparedStatement ps;
    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("init");
        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false", "root", "root");
        ps = conn.prepareStatement("insert into zj_user_space (room_id, building_id, building_name, seat_used, room_join_id, user_ids, campus, floor_area, room_area, dept_name, building_area, `type`, floor_name, room_name, seat_count, user_names, admin_id, floor_id, `state`, admin_name, dept_id)\n" +
                "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);");
    }

    @Override
    public void invoke(JSONArray value, Context context) throws Exception {
        if (value.size() != 0){
            List<UserToSpace> userToSpaces = value.toJavaList(UserToSpace.class);

            for (UserToSpace userToSpace : userToSpaces) {
                ps.setString(1, userToSpace.getRoom_id());
                ps.setString(2, userToSpace.getBuilding_id());
                ps.setString(3, userToSpace.getBuilding_name());
                ps.setString(4, userToSpace.getSeat_used());
                ps.setString(5, userToSpace.getRoom_join_id());
                ps.setString(6, userToSpace.getUser_ids());
                ps.setString(7, userToSpace.getCampus());
                ps.setString(8, userToSpace.getFloor_area());
                ps.setString(9, userToSpace.getRoom_area());
                ps.setString(10, userToSpace.getDept_name());
                ps.setString(11, userToSpace.getBuilding_area());
                ps.setString(12, userToSpace.getType());
                ps.setString(13, userToSpace.getFloor_name());
                ps.setString(14, userToSpace.getRoom_name());
                ps.setString(15, userToSpace.getSeat_count());
                ps.setString(16, userToSpace.getUser_names());
                ps.setString(17, userToSpace.getAdmin_id());
                ps.setString(18, userToSpace.getFloor_id());
                ps.setString(19, userToSpace.getState());
                ps.setString(20, userToSpace.getAdmin_name());
                ps.setString(21, userToSpace.getDept_id());
                ps.execute();
            }
        }
    }

    @Override
    public void close() throws Exception {
        ps.close();
        conn.close();
    }
}
